package com.tech

import com.tech.config.ApplicationConfig
import org.apache.kudu.spark.kudu.{KuduContext, KuduDataFrameReader}
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.{SparkConf, SparkContext}

import java.sql.{Connection, DriverManager}
import java.text.SimpleDateFormat
import scala.collection.mutable.ArrayBuffer

object DeleteExpiredPeriodActiveProcess {


  /**
   * 获取活跃活动
   *
   * @return 活跃活动ID
   */
  def getPeriodActiveID(connectionURL: String): ArrayBuffer[String] = {
    val connection: Connection = DriverManager.getConnection(connectionURL)
    val statement = connection.createStatement()
    val set = statement.executeQuery(
      """SELECT id
        |FROM shuat_socialhub.tsr_marketing_activity
        |WHERE (status = 100000002
        |	AND activity_type = 1
        |	AND current_timestamp() BETWEEN begin_time AND end_time)
        |	OR (activity_type = 1
        |		AND date_format(end_time, '%Y%m%d') = date_format(date_sub(curdate(), INTERVAL 1 DAY), '%Y%m%d'))""".stripMargin)
    val array = new ArrayBuffer[String]()
    while (set.next()) {
      array.append(set.getString("id"))
    }
    set.close()
    statement.close()
    connection.close()
    array
  }

  private def getTableData(sparkSession: SparkSession, kuduMaster: String, tableName: String, sql: String): DataFrame = {
    //定义 map 集合，封装 kudu 的 master 地址和要读取的表名
    val options = Map(
      "kudu.master" -> kuduMaster,
      "kudu.table" -> tableName
    )
    val dataFrame: DataFrame = sparkSession.read.options(options).kudu
    dataFrame.createOrReplaceTempView(tableName)
    sparkSession.sql(sql)
  }

  private def deleteData(sparkSession: SparkSession, sc: SparkContext, kuduMaster: String,
                         kuduContext: KuduContext, tableName: String, sql: String): Unit = {
    val result: DataFrame = getTableData(sparkSession, kuduMaster, tableName, sql)
    //删除对应的数据，这里必须要是主键字段
    kuduContext.deleteRows(result, tableName)
  }

  private def backupData(sparkSession: SparkSession, sc: SparkContext, kuduMaster: String,
                         kuduContext: KuduContext, tableName: String, sql: String): Unit = {
    val result: DataFrame = getTableData(sparkSession, kuduMaster, tableName, sql)
    //删除对应的数据，这里必须要是主键字段
    kuduContext.upsertRows(result, tableName + "_history")
  }

  def main(args: Array[String]): Unit = {
    // 查询mysql获取周期活动id
    val IdCondition = DeleteExpiredPeriodActiveProcess.getPeriodActiveID(ApplicationConfig.URL_MYSQL).mkString("'", "','", "'")

    val sparkConf: SparkConf = new SparkConf().setAppName("DeleteExpiredPeriodActiveProcess").setMaster("local[2]")
    val sparkSession: SparkSession = SparkSession.builder().config(sparkConf).getOrCreate()
    val sc: SparkContext = sparkSession.sparkContext
    sc.setLogLevel("warn")
    //指定 kudu 的 master 地址
    val kuduMaster = ApplicationConfig.KUDU_MASTER
    //构建 KuduContext 对象
    val kuduContext = new KuduContext(kuduMaster, sc)

    val strDate = new SimpleDateFormat("yyyy-MM-dd").format(System.currentTimeMillis())
    val sqlCondition = s"where activity_id in ($IdCondition) and create_time< cast('$strDate' as Timestamp)"


    // 删除三张表, 分别构建各自的sql语句
    var tableName = "customer_activity"
    var sql = s"select * from $tableName $sqlCondition"
    backupData(sparkSession, sc, kuduMaster, kuduContext, tableName, sql)
    sql = s"select customer_id,activity_id from $tableName $sqlCondition"
    deleteData(sparkSession, sc, kuduMaster, kuduContext, tableName, sql)

    tableName = "customer_node"
    sql = s"select * from $tableName $sqlCondition"
    backupData(sparkSession, sc, kuduMaster, kuduContext, tableName, sql)
    sql = s"select customer_id,activity_id,process_id from $tableName $sqlCondition"
    deleteData(sparkSession, sc, kuduMaster, kuduContext, tableName, sql)

    tableName = "customer_node_behavior"
    sql = s"select customer_id,activity_id,process_id from $tableName $sqlCondition"
    deleteData(sparkSession, sc, kuduMaster, kuduContext, tableName, sql)

    tableName = "delay_node"
    sql = s"select project,customer_id,activity_id,process_id from $tableName $sqlCondition"
    deleteData(sparkSession, sc, kuduMaster, kuduContext, tableName, sql)

    sparkSession.close()
  }

}
